package org.apache.flink.runtime.metrics.dump;

/** Container for scope related information as required by the MetricQueryService. */
public abstract class QueryScopeInfo {
    /**
     * Categories to be returned by {@link QueryScopeInfo#getCategory()} to avoid instanceof checks.
     */
    /** */
    public static final byte INFO_CATEGORY_CONSUMER_MANAGER = 0;
    public static final byte INFO_CATEGORY_RESOUCE_PROVIDER = 1;
    public static final byte INFO_CATEGORY_CONSUMER = 2;
    public static final byte INFO_CATEGORY_RESOURCESLOT = 3;

    /** The remaining scope not covered by specific fields. */
    public final String scope;

    private QueryScopeInfo(String scope) {
        this.scope = scope;
    }

    /**
     * Create a copy of this QueryScopeInfo and append the given scope.
     *
     * @param userScope scope to append
     * @return modified copy of this QueryScopeInfo
     */
    public abstract QueryScopeInfo copy(String userScope);

    /**
     * Returns the category for this QueryScopeInfo.
     *
     * @return category
     */
    public abstract byte getCategory();
    

    @Override
    public String toString() {
        return "QueryScopeInfo{"
                + "scope='"
                + scope
                + '\''
                + ", category='"
                + getCategory()
                + '\''
                + '}';
    }

    protected String concatScopes(String additionalScope) {
        return scope.isEmpty() ? additionalScope : scope + "." + additionalScope;
    }

    /** Container for the consumer manager scope. Stores no additional information. */
    public static class ConsumerManagerQueryScopeInfo extends QueryScopeInfo {
        public ConsumerManagerQueryScopeInfo() {
            super("");
        }

        public ConsumerManagerQueryScopeInfo(String scope) {
            super(scope);
        }

        @Override
        public ConsumerManagerQueryScopeInfo copy(String additionalScope) {
            return new ConsumerManagerQueryScopeInfo(concatScopes(additionalScope));
        }

        @Override
        public byte getCategory() {
            return INFO_CATEGORY_CONSUMER_MANAGER;
        }
    }

    /** Container for the task manager scope. Stores the ID of the task manager. */
    public static class ResourceProviderQueryScopeInfo extends QueryScopeInfo {
        public final String resourceProviderID;

        public ResourceProviderQueryScopeInfo(String taskManagerId) {
            this(taskManagerId, "");
        }

        public ResourceProviderQueryScopeInfo(String taskManagerId, String scope) {
            super(scope);
            this.resourceProviderID = taskManagerId;
        }

        @Override
        public ResourceProviderQueryScopeInfo copy(String additionalScope) {
            return new ResourceProviderQueryScopeInfo(this.resourceProviderID, concatScopes(additionalScope));
        }

        @Override
        public byte getCategory() {
            return INFO_CATEGORY_RESOUCE_PROVIDER;
        }
    }

    /** Container for the job scope. Stores the ID of the job. */
    public static class ConsumerQueryScopeInfo extends QueryScopeInfo {
        public final String consumerID;

        public ConsumerQueryScopeInfo(String jobID) {
            this(jobID, "");
        }

        public ConsumerQueryScopeInfo(String jobID, String scope) {
            super(scope);
            this.consumerID = jobID;
        }

        @Override
        public ConsumerQueryScopeInfo copy(String additionalScope) {
            return new ConsumerQueryScopeInfo(this.consumerID, concatScopes(additionalScope));
        }

        @Override
        public byte getCategory() {
            return INFO_CATEGORY_CONSUMER;
        }
    }

    /** Container for the resource(slot) scope. Stores the ID of the provider/slot. */
    public static class ResourceSlotQueryScopeInfo extends QueryScopeInfo {
        public final String providerID;
        public final String slotID;
        public final int attemptNumber;

        public ResourceSlotQueryScopeInfo(
                String jobID, String vertexid, int attemptNumber) {
            this(jobID, vertexid, attemptNumber, "");
        }

        public ResourceSlotQueryScopeInfo(
                String jobID, String vertexid, int attemptNumber, String scope) {
            super(scope);
            this.providerID = jobID;
            this.slotID = vertexid;
            this.attemptNumber = attemptNumber;
        }

        @Override
        public ResourceSlotQueryScopeInfo copy(String additionalScope) {
            return new ResourceSlotQueryScopeInfo(
                    this.providerID,
                    this.slotID,
                    this.attemptNumber,
                    concatScopes(additionalScope));
        }

        @Override
        public byte getCategory() {
            return INFO_CATEGORY_RESOURCESLOT;
        }
    }
}
